#ifdef CONFIG_MSPROTOCOL_SRT
#define SRT_C
/*----------------------------------------------*
 * include files            	 ������ͷ�ļ�   *
 *----------------------------------------------*/
#if defined(OS_ANDROID)||defined(OS_LINUX_SOC)
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <err.h>
#include <linux/tcp.h>
#elif defined(OS_WIN32)
#include <winsock2.h>
#include <ws2tcpip.h>
#endif
#include <srt/srt.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
//user header
#include <libmslog/mslog.h>
#include <libmscommon/mscommon.h>
#include <libmscommon/msstring.h>
#include <libmscommon/msnetwork.h>
#include <libmscommon/msepoll.h>
#include <libmscommon/mstime.h>
#include "msprotocol.h"


 #define FLAG "SRT"
typedef enum SRTMode {
    SRT_MODE_CALLER = 0,
    SRT_MODE_LISTENER = 1,
    SRT_MODE_RENDEZVOUS = 2
}SRTMode;
typedef struct MSSRTContext {
	ms_s32 listen_fd;	//tcp socket use
	ms_s32 msepoll_opt;
	ms_s32 msepoll_fd;
	SRT_MSGCTRL mctrl;
//SRT参数设置
	ms_s64 maxbw;
	ms_s32 pbkeylen;
	ms_byte passphrase[256];
#if SRT_VERSION_VALUE >= 0x010302
	ms_s32 enforced_encryption;
	ms_s32 kmrefreshrate;
	ms_s32 kmpreannounce;
#endif
	ms_s32 mss;
	ms_s32 ffs;
	ms_s32 ipttl;
	ms_s32 iptos;
	ms_s64 inputbw;
	ms_s32 oheadbw;
	ms_s64 latency;
	ms_s32 tsbpdMode;
	ms_s32 tlpktdrop;
	ms_s32 nakreport;
	ms_s64 connect_timeout;
	ms_s32 payload_size;
	ms_s64 rcvlatency;
	ms_s64 peerlatency;
	SRTMode mode;
	ms_s32 sndbuf;
	ms_s32 rcvbuf;
	ms_s32 lossmaxttl;
	ms_s32 minversion;
	ms_byte streamid[256];
	ms_byte smoother[256];
	ms_s32 messageapi;
	SRT_TRANSTYPE transtype;
	ms_s32 linger;
} MSSRTContext;

#define mssrtepoll_timeout_ms		500
static int mssrt_close(URLContext *s);
static void mssrt_optinit(URLContext *s){
	//maybe change by user		
	OPTVAL_INIT("rw_timeout",s->opt.srt.rw_timeout, 0, 10);	//Note:client's timeout cannot less than rw_timeout
	OPTVAL_INIT("send_buffer_size",s->opt.srt.send_buffer_size, 0, -1);
	OPTVAL_INIT("recv_buffer_size",s->opt.srt.recv_buffer_size, 0, -1);		
	//no need to change	
	OPTVAL_INIT("block",s->opt.srt.block, 0, 0);		
	OPTVAL_INIT("reuse",s->opt.srt.reuse, 0, 1);			
	OPTVAL_INIT("backlog",s->opt.srt.backlog, 0,10);			
} 
static int mssrt_epoll_open(void )
{	
	int ret=srt_epoll_create();
	if(ret<0){
		ms_error("srt_epoll_create is failed,(%s)",srt_getlasterror_str());
	}
	return ret;
}
static int mssrt_epoll_add(int epfd, int ev_fd, const int* opt)
{
	int ret=srt_epoll_add_usock(epfd, (SRTSOCKET)ev_fd,opt);
	if(ret<0){
		ms_error("srt_epoll_add_usock is failed,(%s)(epfd, ev_fd:%d,%d)",srt_getlasterror_str(),epfd, (SRTSOCKET)ev_fd);
	}
	return ret;
}

static int mssrt_epoll_mod(int epfd, int ev_fd, const int* opt)
{
	int ret=srt_epoll_update_usock(epfd, (SRTSOCKET)ev_fd,opt);
	if(ret<0){
		ms_error("srt_epoll_update_usock is failed,(%s)",srt_getlasterror_str());
	}
	return ret;
}
int mssrt_epoll_del(int epfd, int ev_fd)
{
	int ret=srt_epoll_remove_usock(epfd, (SRTSOCKET)ev_fd);
	if(ret<0){
		ms_error("srt_epoll_remove_usock is failed,(%s)",srt_getlasterror_str());
	}
	return ret;
}
static int mssrt_epoll_waitTimeout(int epfd, ms_bool is_out,int timeout_ms)
{
    int ret, len = 1, errlen = 1;
    SRTSOCKET ready[1];
    SRTSOCKET error[1];

    if (ms_true==is_out) {
        ret = srt_epoll_wait(epfd, error, &errlen, ready, &len, timeout_ms, 0, 0, 0, 0);
    } else {
        ret = srt_epoll_wait(epfd, ready, &len, error, &errlen, timeout_ms, 0, 0, 0, 0);
    }
	//ms_debug("ret=%d,len=%d,errlen=%d",ret,len,errlen);
    if (!(ret < 0)) {
        ret = errlen ? -1 : 0;
    }
    return ret;
}
static int mssrt_epoll_waitfdTimeout(int epfd,int ev_fd, ms_bool is_out,int timeout_ms)
{
	int event_index=0;
	int opt= ((ms_true==is_out) ? SRT_EPOLL_OUT : SRT_EPOLL_IN)|SRT_EPOLL_ERR;
	mssrt_epoll_add(epfd,ev_fd,&opt);
	int ret=mssrt_epoll_waitTimeout(epfd, is_out,timeout_ms);
	mssrt_epoll_del(epfd,ev_fd);
	return ret;
}
static int mssrt_epoll_close(int epfd)
{
	return srt_epoll_release(epfd);
}

static int mssrt_epoll_waitfdTimeout2(int ev_fd, ms_bool is_out,int timeout_ms)
{
	int event_index=0;
	ms_s32 epfd = mssrt_epoll_open();
	int ret=mssrt_epoll_waitfdTimeout(epfd, ev_fd,is_out,timeout_ms);
	mssrt_epoll_close(epfd);
	return ret;
}


void mssrt_init()
{
	ms_u32 srtver=srt_getversion();
	ms_info("srt versiob:%d.%d.%d",((srtver>>16)&0xFF),((srtver>>8)&0xFF),(srtver&0xFF));
	int ret=srt_startup();
	if(-1==ret){
		ms_error("srt_startup failed!");
	}else if(0==ret){
		ms_info("Successfully run, or already started");
	}else if(1==ret){
		ms_waring("This is the first startup, but the GC thread is already running");
	}
}
void mssrt_deinit()
{
	srt_cleanup();
}

ms_s32 mssrt_getSockState(int srt_fd )
{	
	return srt_getsockstate(srt_fd);
}
static ms_s08 mssrt_setsockopt(int fd, SRT_SOCKOPT optname, const char * optnamestr, const void * optval, int optlen)
{
	if (srt_setsockopt(fd, 0, optname, optval, optlen) < 0) {
		ms_errRet(-1, "failed to set option %s on socket: %s", optnamestr, srt_getlasterror_str());
	}
	return 0;
}
static ms_s08 mssrt_getsockopt(int fd, SRT_SOCKOPT optname, const char * optnamestr, void * optval, int * optlen)
{
    if (srt_getsockopt(fd, 0, optname, optval, optlen) < 0) {
        ms_errRet(-1, "failed to get option %s on socket: %s", optnamestr, srt_getlasterror_str());
    }
    return 0;
}

static ms_bool mssrt_connect(URLContext *s,SRTSOCKET srt_fd,struct sockaddr_in sa)
{
	ms_bool ret=ms_true;
	ms_debug("try to srt-connect");
	ms_s32 st = srt_connect(srt_fd, (struct sockaddr*)&sa, sizeof sa);
	if (st == SRT_ERROR){
		ms_error("srt_connect failed: %s", srt_getlasterror_str());
		return ms_false;
	}
	if(0==s->opt.srt.block){
		ms_s32 ret_check=mssrt_epoll_waitfdTimeout2(srt_fd, ms_true, s->opt.srt.rw_timeout*100);
		if (ret_check < 0){
			ms_error("srt connect failed: %s", srt_getlasterror_str());
			ret=ms_false;
		}else{
			ms_debug("srt connect success");
		}
	}
	return ret;
}

static int mssrt_socket_nonblock(int socket, int block)
{
    int ret, blocking = block ? 1 : 0;
	mssrt_setsockopt(socket, SRTO_RCVSYN ,  "SRTO_RCVSYN", &blocking, sizeof(blocking));
	mssrt_setsockopt(socket, SRTO_SNDSYN ,  "SRTO_SNDSYN", &blocking, sizeof(blocking));
}
static void mssrt_initOptConfig(MSSRTContext * srt_ctt){
	srt_ctt->maxbw=-1;
	srt_ctt->pbkeylen=-1;
	ms_memset0(srt_ctt->passphrase, ms_bufsize(srt_ctt->passphrase));
#if SRT_VERSION_VALUE >= 0x010302
	srt_ctt->enforced_encryption=-1;
	srt_ctt->kmrefreshrate=-1;
	srt_ctt->kmpreannounce=-1;
#endif
	srt_ctt->mss=-1;
	srt_ctt->ffs=-1;
	srt_ctt->ipttl=-1;
	srt_ctt->iptos=-1;
	srt_ctt->inputbw=-1;
	srt_ctt->oheadbw=-1;
	srt_ctt->latency=-1;
	srt_ctt->tsbpdMode=1;
	srt_ctt->tlpktdrop=1;
	srt_ctt->nakreport=-1;
	srt_ctt->connect_timeout=-1;
	srt_ctt->payload_size=-1;
	srt_ctt->rcvlatency=-1;
	srt_ctt->peerlatency=-1;
	srt_ctt->mode= SRT_MODE_CALLER;
	srt_ctt->sndbuf=-1;
	srt_ctt->rcvbuf=-1;
	srt_ctt->lossmaxttl= -1;
	srt_ctt->minversion= -1;
	ms_memset0(srt_ctt->streamid, ms_bufsize(srt_ctt->streamid));
	ms_memset0(srt_ctt->smoother, ms_bufsize(srt_ctt->smoother));
	srt_ctt->messageapi= -1;
	srt_ctt->transtype= SRTT_INVALID;
	srt_ctt->linger= -1;
};

static void mssrt_getOptConfig(URLContext *s)
{
	MSSRTContext * srt_ctt=s->priv_data;
	ms_byte buf[256]={0};
	ms_byte *p = strchr(s->url, '?');
	if (p) {
		if (msstr_api_findTag(buf, sizeof(buf), "maxbw", p)) {
			srt_ctt->maxbw = strtoll(buf, ms_null, 0);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "pbkeylen", p)) {
			srt_ctt->pbkeylen = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "passphrase", p)) {
			ms_strcpy(srt_ctt->passphrase, buf);
		}
#if SRT_VERSION_VALUE >= 0x010302
		if (msstr_api_findTag(buf, sizeof(buf), "enforced_encryption", p)) {
			srt_ctt->enforced_encryption = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "kmrefreshrate", p)) {
			srt_ctt->kmrefreshrate = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "kmpreannounce", p)) {
			srt_ctt->kmpreannounce = strtol(buf, ms_null, 10);
		}
#endif
		if (msstr_api_findTag(buf, sizeof(buf), "mss", p)) {
			srt_ctt->mss = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "ffs", p)) {
			srt_ctt->ffs = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "ipttl", p)) {
			srt_ctt->ipttl = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "iptos", p)) {
			srt_ctt->iptos = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "inputbw", p)) {
			srt_ctt->inputbw = strtoll(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "oheadbw", p)) {
			srt_ctt->oheadbw = strtoll(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "latency", p)) {
			srt_ctt->latency = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "tsbpddelay", p)) {
			srt_ctt->latency = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "rcvlatency", p)) {
			srt_ctt->rcvlatency = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "peerlatency", p)) {
			srt_ctt->peerlatency = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "tsbpd", p)) {
			srt_ctt->tsbpdMode = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "tlpktdrop", p)) {
			srt_ctt->tlpktdrop = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "nakreport", p)) {
			srt_ctt->nakreport = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "connect_timeout", p)) {
			srt_ctt->connect_timeout = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "payload_size", p) ||
			msstr_api_findTag(buf, sizeof(buf), "pkt_size", p)) {
			srt_ctt->payload_size = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "mode", p)) {
			if (!strcmp(buf, "listener")) {
				srt_ctt->mode = SRT_MODE_LISTENER;
				s->opt.srt.listen=1;
			}else if (!strcmp(buf, "rendezvous")) {
				srt_ctt->mode = SRT_MODE_RENDEZVOUS;
			}else{//默认是CALLER模式
				srt_ctt->mode = SRT_MODE_CALLER;
			}
		}
		if (msstr_api_findTag(buf, sizeof(buf), "sndbuf", p)) {
			srt_ctt->sndbuf = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "rcvbuf", p)) {
			srt_ctt->rcvbuf = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "lossmaxttl", p)) {
			srt_ctt->lossmaxttl = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "minversion", p)) {
			srt_ctt->minversion = strtol(buf, ms_null, 0);
		}
		//srt://127.0.0.1:10080?streamid=#!::r=live/livestream,m=publish
		if (msstr_api_findTag(buf, sizeof(buf), "streamid", p)) {
			ms_strcpy(srt_ctt->streamid, buf);
			ms_debug("find streamid:%s",srt_ctt->streamid);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "smoother", p)) {
			ms_strcpy(srt_ctt->smoother, buf);
			ms_debug("find smoother:%s",srt_ctt->smoother);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "messageapi", p)) {
			srt_ctt->messageapi = strtol(buf, ms_null, 10);
		}
		if (msstr_api_findTag(buf, sizeof(buf), "transtype", p)) {
			if (!strcmp(buf, "file")) {
				srt_ctt->transtype = SRTT_FILE;
			}else{
				srt_ctt->transtype = SRTT_LIVE;
			}
		}
		if (msstr_api_findTag(buf, sizeof(buf), "linger", p)) {
			srt_ctt->linger = strtol(buf, ms_null, 10);
		}
		
	}
}
static ms_s08 mssrt_setOptConfig(URLContext *s,SRTSOCKET *psrt_fd)
{
	MSSRTContext * srt_ctt=s->priv_data;
    int yes = 1;
    int latency = srt_ctt->latency / 1000;
    int rcvlatency = srt_ctt->rcvlatency / 1000;
    int peerlatency = srt_ctt->peerlatency / 1000;
    int connect_timeout = srt_ctt->connect_timeout;

	if ((srt_ctt->mode == SRT_MODE_RENDEZVOUS && mssrt_setsockopt((*psrt_fd), SRTO_RENDEZVOUS, "SRTO_RENDEZVOUS", &yes, sizeof(yes)) < 0) ||
		(srt_ctt->transtype != SRTT_INVALID && mssrt_setsockopt((*psrt_fd), SRTO_TRANSTYPE, "SRTO_TRANSTYPE", &srt_ctt->transtype, sizeof(srt_ctt->transtype)) < 0) ||
		(srt_ctt->maxbw >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_MAXBW, "SRTO_MAXBW", &srt_ctt->maxbw, sizeof(srt_ctt->maxbw)) < 0) ||
		(srt_ctt->pbkeylen >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_PBKEYLEN, "SRTO_PBKEYLEN", &srt_ctt->pbkeylen, sizeof(srt_ctt->pbkeylen)) < 0) ||
		(ms_buflen(srt_ctt->passphrase)>0 && mssrt_setsockopt((*psrt_fd), SRTO_PASSPHRASE, "SRTO_PASSPHRASE", srt_ctt->passphrase, strlen(srt_ctt->passphrase)) < 0) ||
#if SRT_VERSION_VALUE >= 0x010302
#if SRT_VERSION_VALUE >= 0x010401
		(srt_ctt->enforced_encryption >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_ENFORCEDENCRYPTION, "SRTO_ENFORCEDENCRYPTION", &srt_ctt->enforced_encryption, sizeof(srt_ctt->enforced_encryption)) < 0) ||
#else
		/* SRTO_STRICTENC == SRTO_ENFORCEDENCRYPTION (53), but for compatibility, we used SRTO_STRICTENC */
		(srt_ctt->enforced_encryption >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_STRICTENC, "SRTO_STRICTENC", &srt_ctt->enforced_encryption, sizeof(srt_ctt->enforced_encryption)) < 0) ||
#endif
		(srt_ctt->kmrefreshrate >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_KMREFRESHRATE, "SRTO_KMREFRESHRATE", &srt_ctt->kmrefreshrate, sizeof(srt_ctt->kmrefreshrate)) < 0) ||
		(srt_ctt->kmpreannounce >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_KMPREANNOUNCE, "SRTO_KMPREANNOUNCE", &srt_ctt->kmpreannounce, sizeof(srt_ctt->kmpreannounce)) < 0) ||
#endif
		(srt_ctt->mss >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_MSS, "SRTO_MSS", &srt_ctt->mss, sizeof(srt_ctt->mss)) < 0) ||
		(srt_ctt->ffs >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_FC, "SRTO_FC", &srt_ctt->ffs, sizeof(srt_ctt->ffs)) < 0) ||
		(srt_ctt->ipttl >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_IPTTL, "SRTO_IPTTL", &srt_ctt->ipttl, sizeof(srt_ctt->ipttl)) < 0) ||
		(srt_ctt->iptos >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_IPTOS, "SRTO_IPTOS", &srt_ctt->iptos, sizeof(srt_ctt->iptos)) < 0) ||
		(srt_ctt->latency >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_LATENCY, "SRTO_LATENCY", &latency, sizeof(latency)) < 0) ||
		(srt_ctt->rcvlatency >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_RCVLATENCY, "SRTO_RCVLATENCY", &rcvlatency, sizeof(rcvlatency)) < 0) ||
		(srt_ctt->peerlatency >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_PEERLATENCY, "SRTO_PEERLATENCY", &peerlatency, sizeof(peerlatency)) < 0) ||
		(srt_ctt->tsbpdMode >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_TSBPDMODE, "SRTO_TSBPDMODE", &srt_ctt->tsbpdMode, sizeof(srt_ctt->tsbpdMode)) < 0) ||
		(srt_ctt->tlpktdrop >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_TLPKTDROP, "SRTO_TLPKTDROP", &srt_ctt->tlpktdrop, sizeof(srt_ctt->tlpktdrop)) < 0) ||
		(srt_ctt->nakreport >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_NAKREPORT, "SRTO_NAKREPORT", &srt_ctt->nakreport, sizeof(srt_ctt->nakreport)) < 0) ||
		(connect_timeout >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_CONNTIMEO, "SRTO_CONNTIMEO", &connect_timeout, sizeof(connect_timeout)) <0 ) ||
		(srt_ctt->sndbuf >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_SNDBUF, "SRTO_SNDBUF", &srt_ctt->sndbuf, sizeof(srt_ctt->sndbuf)) < 0) ||
		(srt_ctt->rcvbuf >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_RCVBUF, "SRTO_RCVBUF", &srt_ctt->rcvbuf, sizeof(srt_ctt->rcvbuf)) < 0) ||
		(srt_ctt->lossmaxttl >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_LOSSMAXTTL, "SRTO_LOSSMAXTTL", &srt_ctt->lossmaxttl, sizeof(srt_ctt->lossmaxttl)) < 0) ||
		(srt_ctt->minversion >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_MINVERSION, "SRTO_MINVERSION", &srt_ctt->minversion, sizeof(srt_ctt->minversion)) < 0) ||
		(ms_buflen(srt_ctt->streamid)>0&& mssrt_setsockopt((*psrt_fd), SRTO_STREAMID, "SRTO_STREAMID", srt_ctt->streamid, strlen(srt_ctt->streamid)) < 0) ||
#if SRT_VERSION_VALUE >= 0x010401
		(ms_buflen(srt_ctt->smoother)>0&& mssrt_setsockopt((*psrt_fd), SRTO_CONGESTION, "SRTO_CONGESTION", srt_ctt->smoother, strlen(srt_ctt->smoother)) < 0) ||
#else
		(ms_buflen(srt_ctt->smoother)>0&& mssrt_setsockopt((*psrt_fd), SRTO_SMOOTHER, "SRTO_SMOOTHER", srt_ctt->smoother, strlen(srt_ctt->smoother)) < 0) ||
#endif
		(srt_ctt->messageapi >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_MESSAGEAPI, "SRTO_MESSAGEAPI", &srt_ctt->messageapi, sizeof(srt_ctt->messageapi)) < 0) ||
		(srt_ctt->payload_size >= 0 && mssrt_setsockopt((*psrt_fd), SRTO_PAYLOADSIZE, "SRTO_PAYLOADSIZE", &srt_ctt->payload_size, sizeof(srt_ctt->payload_size)) < 0) ||
		((s->flags&FLAG_WRITE) && mssrt_setsockopt((*psrt_fd), SRTO_SENDER, "SRTO_SENDER", &yes, sizeof(yes)) < 0)) {
		return -1;
	}
	
	if (srt_ctt->linger >= 0) {
		struct linger lin;
		lin.l_linger = srt_ctt->linger;
		lin.l_onoff  = lin.l_linger > 0 ? 1 : 0;
		if (mssrt_setsockopt((*psrt_fd), SRTO_LINGER, "SRTO_LINGER", &lin, sizeof(lin)) < 0)
			return -1;
	}
	return 0;

}
static int mssrt_accepts(URLContext *s);
static void srtserver_accecpt(void *arg)
{
	URLContext *s=arg;
	MSSRTContext * srt_ctt=s->priv_data;
	SRTSERVERContext *pSrtServer_ctt=&s->opt.srt.srtServer_ctt;
	int i=0;
	while(ms_true){
		if(pSrtServer_ctt->clientNum<CONFIG_MAX_SRTCLIENT){
			if (!(mssrt_epoll_waitTimeout(srt_ctt->msepoll_fd, ms_false,s->opt.srt.rw_timeout*100)< 0)){
				int fd=mssrt_accepts(s);
				if(fd>0){
					ms_debug("accept success");
					for( i=0;i<CONFIG_MAX_SRTCLIENT;i++){
						if(ms_true!=pSrtServer_ctt->flag_used[i]){
							pSrtServer_ctt->clientFd[i]=fd;
							pSrtServer_ctt->flag_used[i]=ms_true;
							pSrtServer_ctt->clientNum+=1;
							break;
						}
					}
				}
			}
		}
		if(ms_true==pSrtServer_ctt->flag_stopAccecpt){
			for( i=0;i<CONFIG_MAX_SRTCLIENT;i++){
				if(ms_true==pSrtServer_ctt->flag_used[i]){
					srt_close(pSrtServer_ctt->clientFd[i]);
					pSrtServer_ctt->flag_used[i]=ms_false;
					pSrtServer_ctt->clientNum-=1;
				}
			}
			pSrtServer_ctt->flag_stopAccecpt=ms_false;
			return;
		}
		ms_msleep(100);
	}
}

static int mssrt_open(URLContext *s)
{
	if(1==s->is_connected){
		return 0;
	}
	mssrt_optinit(s);
	
	msptc_api_urlSplit_urlctt(s);
	
	MSSRTContext * srt_ctt=s->priv_data=(void *)ms_mallocDes(sizeof(MSSRTContext),"MSSRTContext",s->url);
	if(ms_null==s->priv_data){
		ms_errGoto(msfail, "ms_malloc MSSRTContext faild(%d,%s)",errno,strerror(errno));
	}
	ms_memset(srt_ctt,0,sizeof(MSSRTContext));

	mssrt_initOptConfig(srt_ctt);
	ms_verbose("===========Open %s,flags:%d",s->url,s->flags);
	mssrt_getOptConfig(s);

	/*Init add info*/
	SRTSOCKET srt_fd= srt_create_socket();
	if (srt_fd <0){
		ms_errRet(-1, "srt_create_socket failed!Error: %s", srt_getlasterror_str());
	}
	struct sockaddr_in sa;
    sa.sin_family = AF_INET;
    sa.sin_port = htons( s->url_info.port);
    if (inet_pton(AF_INET, s->url_info.hostname, &sa.sin_addr) != 1){
		ms_errGoto(msfail,"inet_pton failed: %d,%s", errno, strerror(errno));
    }
	//设置阻塞或非阻塞模式
	mssrt_socket_nonblock(srt_fd, s->opt.srt.block);
		
	mssrt_setOptConfig(s, &srt_fd);
	
	ms_s32 st = 0;
	if(1==s->opt.srt.listen){	
	    int reuse = 1;
		mssrt_setsockopt(srt_fd, SRTO_REUSEADDR, "SRTO_REUSEADDR",&reuse, ms_bufsize(reuse));
		ms_debug("srt bind");
		sa.sin_addr.s_addr = htonl(INADDR_ANY);
    	ms_s32 st = srt_bind(srt_fd, (struct sockaddr*)&sa, ms_bufsize(sa));
		if (st <0){
			ms_errGoto(msfail,"srt_bind: %s", srt_getlasterror_str());
    	}
		/* Create our srt listening socket. */
		ms_debug("Create our srt listening socket(%d)",srt_fd);	
		st=srt_listen(srt_fd, 1);
		if (st <0){
			ms_errGoto(msfail, "srt_listen: %s",srt_getlasterror_str());
		} 
		srt_ctt->listen_fd = srt_fd;
	}else{
		if(SRT_MODE_RENDEZVOUS==srt_ctt->mode){
			struct sockaddr_in local_name;
		    local_name.sin_family = AF_INET;
		    local_name.sin_port = htons( s->url_info.port);
			local_name.sin_addr.s_addr = htonl(INADDR_ANY);
			ms_debug("srt_rendezvous");
		 	st = srt_rendezvous(srt_fd, (const struct sockaddr*)&local_name, ms_bufsize(local_name),(const struct sockaddr*)&sa, ms_bufsize(sa));
			if (st <0){
				ms_errGoto(msfail,"srt_rendezvous: %s", srt_getlasterror_str());
			}
		}else{
			if(ms_false==mssrt_connect( s, srt_fd,sa)){
				ms_errGoto(msfail,"mssrt_connect to %s failed", s->url);
			}
		}
		s->fd = srt_fd;	
	}
 	srt_ctt->msepoll_fd = mssrt_epoll_open();
	if (srt_ctt->msepoll_fd<0) {
		 ms_errGoto(msfail,"mssrt_epoll_open failed");
	 }
	if(1==s->opt.srt.listen){
		if(msthread_api_create(&s->opt.srt.srtServer_ctt.mstd_accecpt, "srt_accecpt",srtserver_accecpt, s)<0){
			ms_debug("msthread_api_create:srt_accecpt failed");
		}
		srt_ctt->msepoll_opt= SRT_EPOLL_IN|SRT_EPOLL_ERR;
	}else{
		srt_ctt->msepoll_opt= ((s->flags&FLAG_WRITE) ? SRT_EPOLL_OUT : SRT_EPOLL_IN)|SRT_EPOLL_ERR;
	}
	mssrt_epoll_add(srt_ctt->msepoll_fd,srt_fd,&srt_ctt->msepoll_opt);
	srt_ctt->mctrl= srt_msgctrl_default;
	s->is_connected=1;
	return 0;
msfail:
	if(ms_null!=srt_ctt){	
		if( srt_ctt->msepoll_fd>0){
			mssrt_epoll_close(srt_ctt->msepoll_fd);
		}
		ms_free(srt_ctt);
		srt_ctt=ms_null;
	}
	if(srt_fd>0){
		st = srt_close(srt_fd);	
		if (st <0){
			ms_error("srt_close: %s", srt_getlasterror_str());
		}
	}
	s->is_connected=0;
	return -1;
}
static int mssrt_accepts(URLContext *s)
{
	if(0==s->is_connected){
		return -1;
	}
	MSSRTContext * srt_ctt=s->priv_data;
	s->peer_info.len=sizeof(s->peer_info.sockaddrs);
	s->fd = srt_accept(srt_ctt->listen_fd, (struct sockaddr *)&s->peer_info.sockaddrs, &s->peer_info.len);
	if (s->fd==SRT_INVALID_SOCK){
		srt_close(s->fd);
		//ms_errRet(-1, "accept error:  %s", srt_getlasterror_str());
		return -1;
	}else{
		ms_debug("srt_accept our srt listening socket(%d),accept_fd(%d)",srt_ctt->listen_fd,s->fd);
	}
#if ENABLE_IPV6
	ms_byte ipaddrbuf[256]={0};
	msnet_api_getIpaddr((struct sockaddr *)&s->peer_info.sockaddrs6,ipaddrbuf);
	ms_sprintfs(s->opt.srt.clientip,"%s",ipaddrbuf);
#else	
	ms_sprintfs(s->opt.srt.clientip,"%s",inet_ntoa(s->peer_info.sockaddrs.sin_addr));
#endif
	ms_verbose("New client from %s,flags:%d(%d---%d)",s->opt.srt.clientip,s->flags,srt_ctt->listen_fd,s->fd);
	if(s->opt.srt.recv_buffer_size>0){
		int old_bufsize	=0;	
		int len=sizeof(old_bufsize);
		srt_getsockopt(s->fd , 0, SRTO_RCVBUF, (char *)&old_bufsize, (socklen_t * __restrict__)&len);
		if(old_bufsize!=s->opt.srt.recv_buffer_size){
			if (srt_setsockopt(s->fd , 0, SRTO_RCVBUF, (const char *)&s->opt.srt.recv_buffer_size, sizeof(s->opt.srt.recv_buffer_size)) < 0) {
				ms_errRet(-1, "setsockopt(SO_RECVBUF) %s(%s)", s->url, srt_getlasterror_str());
			}			
			ms_debug("change SO_RCVBUF form %d to %d",old_bufsize,s->opt.srt.recv_buffer_size);
		}
	}
	if(s->opt.srt.send_buffer_size>0){
		int old_bufsize	=0;	
		int len=sizeof(old_bufsize);
		srt_getsockopt(s->fd , 0, SRTO_SNDBUF, (char *)&old_bufsize, (socklen_t * __restrict__)&len);
		if(old_bufsize!=s->opt.srt.send_buffer_size){
			if (srt_setsockopt(s->fd , 0, SRTO_SNDBUF, (const char *)&s->opt.srt.send_buffer_size, sizeof(s->opt.srt.send_buffer_size)) < 0) {
				ms_errRet(-1, "setsockopt(SO_SNDBUF) %s(%s)", s->url, srt_getlasterror_str());
			}			
			ms_debug("change SO_SNDBUF form %d to %d",old_bufsize,s->opt.srt.send_buffer_size);
		}
	}
	//设置阻塞或非阻塞模式
	mssrt_socket_nonblock(s->fd, s->opt.srt.block);
	return s->fd;	
}

static int mssrt_read(URLContext *s ,  unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}
	MSSRTContext * srt_ctt=s->priv_data;
	int ret=0;	
	if (0==s->opt.srt.block){
		ret=mssrt_epoll_waitTimeout(srt_ctt->msepoll_fd, ms_false, s->opt.srt.rw_timeout*100);
		if (ret < 0){
			return ret;
		}
	}
	ret = srt_recvmsg(s->fd, buf, size);
	if (ret<0){
		mssrt_close(s);
		ms_errRet(-1, "recv error(%d): %s",ret,srt_getlasterror_str());
	}
	return ret;
}
static int mssrt_write(URLContext *s, unsigned char *buf, int size)
{
	if(0==s->is_connected){
		return -1;
	}
	MSSRTContext * srt_ctt=s->priv_data;
	int ret=0;
	if(1==s->opt.srt.listen){
		SRTSERVERContext *pSrtServer_ctt=&s->opt.srt.srtServer_ctt;
		int i=0;
		if(pSrtServer_ctt->clientNum>0){
			int ret_max=0;
			for( i=0;i<CONFIG_MAX_SRTCLIENT;i++){
				if(ms_true==pSrtServer_ctt->flag_used[i]){
					ret = srt_sendmsg(pSrtServer_ctt->clientFd[i], buf, size, -1,0);
					if (ret<0){
						srt_close(pSrtServer_ctt->clientFd[i]);
						pSrtServer_ctt->flag_used[i]=ms_false;
						pSrtServer_ctt->clientNum-=1;
						ms_errRet(-1, "send error(%d): %s",ret,srt_getlasterror_str());
					}
					ret_max=ms_max(ret_max,ret);
				}
			}
			ret=ret_max;
		}else{
			ret= size;
		}
	}else{
		if (0==s->opt.srt.block){
			ret=mssrt_epoll_waitTimeout(srt_ctt->msepoll_fd, ms_true, s->opt.srt.rw_timeout*100);
			if (ret < 0){
				ms_debug("srt mssrt_epoll_waitTimeout,timeout:%d ms",s->opt.srt.rw_timeout*100);
				return ret;
			}
		}
		ret = srt_sendmsg(s->fd, buf, size, -1,0);
		if (ret<0){
			mssrt_close(s);
			ms_errRet(-1, "send error(%d): %s",ret,srt_getlasterror_str());
		}
		return ret;
	}
}
static int mssrt_get_listen_handle(URLContext *s)
{
	if(0==s->is_connected){
		return -1;
	}
	MSSRTContext * srt_ctt=s->priv_data;
	return srt_ctt->listen_fd;
}
static int mssrt_close(URLContext *s)
{
	if(0==s->is_connected){
		return 0;
	}
	MSSRTContext * srt_ctt=s->priv_data;

	if(ms_null!=srt_ctt){
		if(srt_ctt->listen_fd>0){
			SRTSERVERContext *pSrtServer_ctt=&s->opt.srt.srtServer_ctt;
			pSrtServer_ctt->flag_stopAccecpt=ms_true;
			while(ms_true==pSrtServer_ctt->flag_stopAccecpt){
				ms_debug("Wait for srtserver_accecpt stop");
				ms_msleep(100);
			}
			msthread_api_destroy(&pSrtServer_ctt->mstd_accecpt);
			srt_close(srt_ctt->listen_fd);
		}
		if( srt_ctt->msepoll_fd>0){
			if(s->fd>0){
				mssrt_epoll_del(srt_ctt->msepoll_fd, s->fd);
			}
			mssrt_epoll_close(srt_ctt->msepoll_fd);
		}
		ms_free(srt_ctt);
		srt_ctt=ms_null;
	}
	if(s->fd>0){
		srt_close(s->fd);	
		s->fd=-1;
	}
	s->is_connected=0;
	return 0;
}
static ms_void mssrt_bstatsGlobal(SRT_TRACEBSTATS trace,ms_bool flag_sender)
{
	ms_byte bufTmp[256]={0};
	ms_info3("=======全局测量=======");
	mstime_api_ms2str2(trace.msTimeStamp, bufTmp);
	ms_debug3("连接已建立时间:%s",bufTmp);
	if(flag_sender){
		ms_debug3("发送的数据包总数,包括重传:%s(%s),丢失包总数:%d",ms_num2str(trace.pktSentTotal),ms_BYTE64AUTO_unit(trace.byteSentTotal),trace.pktSndLossTotal);	
		ms_debug3("应用程序发送的数据包总数:%s(%s)",ms_num2str(trace.pktSentUniqueTotal),ms_BYTE64AUTO_unit(trace.byteSentUniqueTotal));
		ms_debug3("太晚不能发送的丢包数:%d(%s)",trace.pktSndDropTotal,ms_BYTE64AUTO_unit(trace.byteSndDropTotal));
		ms_debug3("接收ACK报文的总数:%d",trace.pktRecvACKTotal);
		ms_debug3("发送数据的总持续时间(不包括空闲时间):%s",ms_num2str(trace.usSndDurationTotal));	
	}else{
		ms_debug3("接收的数据包总数:%s(%s),丢失包总数:%d",ms_num2str(trace.pktRecvTotal),ms_BYTE64AUTO_unit(trace.byteRecvTotal),trace.pktRcvLossTotal);	
		ms_debug3("应用程序要接收的数据包总数:%s(%s)",ms_num2str(trace.pktRecvUniqueTotal),ms_BYTE64AUTO_unit(trace.byteRecvUniqueTotal)); 
		ms_debug3("太晚不能播发的丢包数:%d(%s)",trace.pktRcvDropTotal,ms_BYTE64AUTO_unit(trace.byteRcvDropTotal));
		ms_debug3("发ACK报文的总数:%d",trace.pktSentACKTotal);
		
	}
	ms_debug3("重传数据包总数:%d(%s)",trace.pktRetransTotal,ms_BYTE64AUTO_unit(trace.byteRetransTotal));	
	ms_debug3("发送/接收NAK报文的总数:%d/%d",trace.pktSentNAKTotal,trace.pktRecvNAKTotal);	
	ms_debug3("未加密报文数:%d(%s)",trace.pktRcvUndecryptTotal,ms_BYTE64AUTO_unit(trace.byteRcvUndecryptTotal));
	ms_debug3("丢失的总字节数:%s",ms_BYTE64AUTO_unit(trace.byteRcvLossTotal));	
}
static ms_void mssrt_bstatsLocal(SRT_TRACEBSTATS trace ,ms_bool flag_sender)
{
	ms_info3("=======本地测量=======");
	if(flag_sender){
		ms_debug3("发送的数据包数,包括重传:%s(%s),丢失包数:%d",ms_num2str(trace.pktSent),ms_BYTE64AUTO_unit(trace.byteSent),trace.pktSndLoss,ms_num2str(trace.byteRecv)); 
		ms_debug3("发送速率:%fMb/s",trace.mbpsSendRate);
		ms_debug3("繁忙发送时间(即不包含空闲时间):%s",ms_num2str(trace.usSndDuration)); 
		ms_debug3("太晚发送的丢包数:%d(%s)",trace.pktSndDrop,ms_BYTE64AUTO_unit(trace.byteSndDrop));  
		ms_debug3("应用程序发送的数据包数:%s(%s)",ms_num2str(trace.pktSentUnique),ms_BYTE64AUTO_unit(trace.byteSentUnique)); 
		ms_debug3("接收ACK报文数:%d/%d",trace.pktRecvACK);	
	}else{
		ms_debug3("接收包数:%s(%s),丢失包数:%d,收到的重传报文数:%d",ms_num2str(trace.pktRecv),ms_BYTE64AUTO_unit(trace.byteRcvLoss)
			,trace.pktRcvLoss,trace.pktRcvRetrans);
		ms_debug3("接收速率:%fMb/s",trace.mbpsRecvRate);	
		ms_debug3("太晚播发的丢包数:%d(%s)",trace.pktRcvDrop,ms_BYTE64AUTO_unit(trace.byteRcvDrop)); 
		ms_debug3("应用程序接收的数据包数:%s(%s)",ms_num2str(trace.pktRecvUnique),ms_BYTE64AUTO_unit(trace.byteRecvUnique)); 
		ms_debug3("发送ACK报文数:%d/%d",trace.pktSentACK);	
	}
	ms_debug3("重传报文数:%d(%s)",trace.pktRetrans,ms_BYTE64AUTO_unit(trace.byteRetrans)); 
	ms_debug3("发送/接收NAK报文数:%d/%d",trace.pktSentNAK,trace.pktRecvNAK);	
	ms_debug3("接收序列的顺序差异大小:%d",trace.pktReorderDistance);	
	ms_debug3("延迟报文(顺序经过ACK的报文)的平均报文延迟时间:%f",trace.pktRcvAvgBelatedTime);	
	ms_debug3("由于来得太晚而被接收和忽略的数据包数:%s",ms_num2str(trace.pktRcvBelated)); 
	ms_debug3("未加密报文数:%d(%s)",trace.pktRcvUndecrypt,ms_BYTE64AUTO_unit(trace.byteRcvUndecrypt)); 
}

ms_void mssrt_bstatsGL(int fd ,ms_bool flag_sender)
{
	SRT_TRACEBSTATS trace;
	srt_bstats(fd, &trace, ms_true);
	mssrt_bstatsGlobal( trace, flag_sender);
	mssrt_bstatsLocal( trace, flag_sender);
}
ms_void mssrt_bstatsInstant(int fd ,ms_bool flag_sender)
{
	SRT_TRACEBSTATS trace;
	srt_bstats(fd, &trace, ms_true);
	ms_info3("=======实时测量=======");
	ms_debug3("报文发送周期:%f毫秒",trace.usPktSndPeriod); 
	ms_debug3("流窗口大小:%d 个数据包",trace.pktFlowWindow);	
	ms_debug3("拥塞窗口大小:%d 个数据包",trace.pktCongestionWindow); 
	ms_debug3("飞行中的包数:%d 个数据包",trace.pktFlightSize);	
	ms_debug3("RTT:%f 毫秒",trace.msRTT);	
	ms_debug3("估计带宽:%f Mb/s",trace.mbpsBandwidth); 
	ms_debug3("可用的发送缓冲区大小:%d",trace.byteAvailSndBuf);
	ms_debug3("可用的接收缓冲区大小:%d",trace.byteAvailRcvBuf);	
	ms_debug3("传输带宽上限:%f Mbps",trace.mbpsMaxBW);	
	ms_debug3("MTU:%d",trace.byteMSS);	
	ms_debug3("发送端UnACK的数据包:%d(%s),时间间隔:%d毫秒 ",trace.byteAvailSndBuf,ms_BYTE32AUTO_unit(trace.byteSndBuf),trace.msSndBuf);
	ms_debug3("基于时间戳的数据包发送延迟:%d",trace.msSndTsbPdDelay);
	ms_debug3("接收器中未发送的数据包:%d(%s),时间间隔:%d毫秒",trace.pktRcvBuf,ms_BYTE32AUTO_unit(trace.byteRcvBuf),trace.msRcvBuf);
	ms_debug3("基于时间戳的数据包发送延迟:%d",trace.msRcvTsbPdDelay);
	ms_debug3("包过滤器提供的控制包数:%d",trace.pktSndFilterExtraTotal);
	ms_debug3("接收到的控制包数,但没有返回:%d",trace.pktRcvFilterExtraTotal);
	ms_debug3("过滤器额外提供的包数(例如FEC重建):%d",trace.pktRcvFilterSupplyTotal);
	ms_debug3("不能被过滤器覆盖的丢包数:%d",trace.pktRcvFilterLossTotal);
	ms_debug3("包过滤器提供的控制包数:%d",trace.pktSndFilterExtra);
	ms_debug3("接收到的控制包数,但没有返回:%d",trace.pktRcvFilterExtra);
	ms_debug3("过滤器额外提供的包数(例如FEC重建):%d",trace.pktRcvFilterSupply);
	ms_debug3("不能被过滤器覆盖的丢包数:%d",trace.pktRcvFilterLoss);
	ms_debug3("包重排序容差值:%d",trace.pktReorderTolerance);
}
URLProtocol ffurl_srt_protocol={
	.nameDes	= msptc_getPtcNameDes(msptc_srt),
    .url_open	= mssrt_open,
    .url_read	= mssrt_read,
    .url_write	= mssrt_write,
    .url_close	= mssrt_close,	
    .priv_func={
		.srt.accept				=mssrt_accepts,
		.srt.get_listen_handle	=mssrt_get_listen_handle,
	},    
    .flags	=STREAM_LIVE,
    .next	=ms_null,
};
#undef SRT_C
#endif
